package com.heima.kafka.simple;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * kafkastream进行流式计算
 */
public class KafkaStreamsStart {

    public static void main(String[] args) {
        /**
         * 2.创建连接配置
         */
        Properties properties=new Properties();
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.104:9092");
        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"applicationId");//必须要加该陪住
        /**
         * 3.创建拓扑图
         */
        StreamsBuilder streamsBuilder=new StreamsBuilder();

        //4.进行流式处理
        streamProcessor(streamsBuilder);


        Topology topology=streamsBuilder.build();
        /**
         *  1.创建kafkastreams对象
         *
         *   KafkaStreams(final Topology topology,
         *                         final Properties props)
         *                         参数1表示拓扑图，表示处理流式计算的业务
         *                         参数2表示连接kafka的配置信息
         */
        KafkaStreams kafkaStreams=new KafkaStreams(topology,properties);

        //5.开始执行kafkastream流式计算
        kafkaStreams.start();
    }

    /**
     * 流式计算的抽取方法
     * @param streamsBuilder
     * 流式处理的目标： 计算单词的个数
     */
    private static void streamProcessor(StreamsBuilder streamsBuilder) {
            //1.接收消息，此时此刻就是消费者的角色
        KStream<String, String> kStream = streamsBuilder.stream("stream-send-topic");
            //2.进行流式处理
        kStream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            /**
             *初始化操作
             * @param value  表示接收到的消息内容   "hello  itcast"   "hello kafka"
             * @return ["hello","itcast","hello","kafka"]
             */
            @Override
            public Iterable<String> apply(String value) {
                return Arrays.asList(value.split(" "));
            }
        })
                /**
                 * 表示分组，key=消息内容的key
                 * value=["hello","itcast","hello","kafka"]
                 */
                .groupBy((key,value)-> value)
                /**
                 * 事件窗口，可以处理晚到的数据，每10s处理一次
                 */
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //求和的api
                .count()
                .toStream()
                        /**
                         * 到此，能够走到当前这步的话，表示流式处理完成
                         * 该map方法表示封装数据流式处理之后的数据，整理要发送给消费者的消息内容
                         */
                .map(new KeyValueMapper<Windowed<String>, Long, KeyValue<String, String>>() {
                        /**
                         * 表示处理消息，准备发送处理之后的消息给kafka
                         * @param key 表示要发送的消息，比如hello    itcast
                         * @param value 表示每个单词出现的次数
                     * @return
                     */
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, Long value) {
                        return new KeyValue<>(key.key(), value.toString());
                    }
                })
                /**
                 * 3.发送消息，指定消费者的topic，此时此刻作为生产者的角色
                 */
                .to("stream_get_topic");

    }
}
